package com.prj.ufdm.demo.tutorial.zkcurator;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
/**
 *  Curator提供了三种Watcher(Cache)来监听结点的变化：
	Path Cache：监视一个路径下1）孩子结点的创建、2）删除，3）以及结点数据的更新。产生的事件会传递给注册的PathChildrenCacheListener。
	Node Cache：监视一个结点的创建、更新、删除，并将结点的数据缓存在本地。
	Tree Cache：Path Cache和Node Cache的“合体”，监视路径下的创建、更新、删除事件，并缓存路径下所有孩子结点的数据。
 * @author Administrator
 *
 */
public class CuratorCatchListen {

	public static void main(String[] args) throws Exception {
		
		String connectString = "47.98.183.17:2181";
		String ZK_PATH = "/zkcurator/zktest";
		
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
		CuratorFramework client = //CuratorFrameworkFactory.newClient(connectString, retryPolicy);
								 CuratorFrameworkFactory.builder()
					                .connectString(connectString)
					                .retryPolicy(retryPolicy)
					                .sessionTimeoutMs(6000)
					                .connectionTimeoutMs(3000)
					                .namespace("demo")
					                .build();
		client.start();
		
		//setNodeCacheListenter(client,ZK_PATH);
		
		//setPathChildrenCacheListenter(client,ZK_PATH);
		
		setTreeCacheListenter(client,ZK_PATH);

        Thread.sleep(Integer.MAX_VALUE);
	}
	
	
	
	/**
	 * Path Children Cache 子节点监听
	 * @param client
	 * @throws Exception
	 */
	private static void setPathChildrenCacheListenter(CuratorFramework client,String path) throws Exception{
        ExecutorService pool = Executors.newCachedThreadPool();
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                System.out.println("开始进行事件分析:-----");
                ChildData data = event.getData();
                switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("CHILD_ADDED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                case CHILD_REMOVED:
                    System.out.println("CHILD_REMOVED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                case CHILD_UPDATED:
                    System.out.println("CHILD_UPDATED : "+ data.getPath() +"  数据:"+ data.getData());
                    break;
                default:
                    break;
                }
            }
        };
        
        PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
        childrenCache.getListenable().addListener(childrenCacheListener);
        System.out.println("Register zk watcher successfully!");
        childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
    }
	
	/**
	 * Node Cache 节点监听
	 * @param client
	 * @throws Exception
	 */
	private static void setNodeCacheListenter(CuratorFramework client,String path) throws Exception{
		 final  NodeCache nodeCache = new NodeCache(client, path, false);
		 NodeCacheListener nodeCacheListener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("the test node is change and result is :");
                System.out.println("path : "+nodeCache.getCurrentData().getPath());
                System.out.println("data : "+new String(nodeCache.getCurrentData().getData()));
                System.out.println("stat : "+nodeCache.getCurrentData().getStat());
            }
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start();
	}
	
	/**
	 * Tree Cache 监听
	 * @param client
	 * @throws Exception
	 */
	private static void setTreeCacheListenter(CuratorFramework client,String path) throws Exception{
		ExecutorService pool = Executors.newCachedThreadPool();
        //设置节点的cache
        TreeCache treeCache = new TreeCache(client, path);
        //设置监听器和处理过程
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if(data !=null){
                    switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println("NODE_ADDED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
                        break;
                    case NODE_REMOVED:
                        System.out.println("NODE_REMOVED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
                        break;
                    case NODE_UPDATED:
                        System.out.println("NODE_UPDATED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
                        break;
                        
                    default:
                        break;
                    }
                }else{
                    System.out.println( "data is null : "+ event.getType());
                }
            }
        });
        //开始监听
        treeCache.start();
	}

}
